//package com.allwees.data.dataport.module.product.mq.consume;
//
//import java.util.List;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Lazy;
//import org.springframework.data.elasticsearch.annotations.Document;
//import org.springframework.stereotype.Component;
//
//import com.allwees.bs.core.model.product.entity.ProductModel;
//import com.allwees.core.mq.rabbitmq.batch.AbstractConsumer;
//import com.allwees.core.mq.rabbitmq.batch.BatchMqConsumer;
//import com.allwees.core.mq.rabbitmq.batch.BatchMqConsumerFactory;
//import com.allwees.data.dataport.module.product.doc.ProductSpu;
//import com.allwees.data.dataport.module.product.repository.es.ProductIndexEsRepository;
//
//import lombok.extern.slf4j.Slf4j;
//
///**
// * 批量接收MQ消息并更新索引（利用{@link BatchMqConsumer}）
// *
// * @author yangylsky
// * @since 2019-11-23
// *
// */
//@Slf4j
//@Component
//@Lazy(false)
//public class BatchUpdateProductIndexListener {
//	private static final String QUEUE_NAME = "q.Product.UpdateIndex";
//
//	@Autowired
//	private BatchMqConsumerFactory batchMqConsumerFactory;
//
//	private BatchMqConsumer<ProductModel, ProductSpu> batchConsumer;
//
//	@Autowired
//	private ProductIndexEsRepository productIndexEsRepository;
//
//	@PostConstruct
//	private void _init() {
//		try {
//			batchConsumer = batchMqConsumerFactory.createConsumer(1000, 2000, QUEUE_NAME);
//			batchConsumer.setMessageConsumer(messageConsumer());
//			batchConsumer.start();
//			log.info(this + " is ready.");
//		} catch(Exception e) {
//			log.error("BatchUpdateProductIndexListener init failed", e);
//		}
//	}
//
//	@PreDestroy
//	private void preDestroy() {
//		batchConsumer.stop();
//	}
//
//	private AbstractConsumer<ProductModel, ProductSpu> messageConsumer() {
//		return new AbstractConsumer<ProductModel, ProductSpu>() {
//			@Override
//			public ProductSpu transform(ProductModel product) {
//				return ProductSpu.fromProductModel(product);
//			}
//
//			@Override
//			public void consume(List<ProductSpu> list) {
//				productIndexEsRepository.saveAll(list);
//			}
//
//			@Override
//			public void afterConsume(int dealNum, long processMillis) {
//				Document docAnno =ProductSpu.class.getAnnotation(Document.class);
//				log.info("Save ProductSpu[num: {}] to index[{}] in {}ms", dealNum, docAnno.indexName(), processMillis);
//			}
//		};
//	}
//
//	@Override
//	public String toString() {
//		return String.format("BatchUpdateProductIndexListener{consumer:" + batchConsumer + "}");
//	}
//}
